package com.ex.kafka.producer;

import com.alibaba.fastjson.JSONObject;
import com.ex.kafka.pojo.User;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Properties;

/**
 * @Author longdeng
 * @Date 2020/3/29 0029 11:02
 */
@Component
public class MyKafkaProducer {

    @Value(value = "zhouxl.kafka.bootstrap.servers")
    private String bootstrapServers;

    @Value(value = "zhouxl.kafka.group.id")
    private String groupId;

    @Value(value = "zhouxl.kafka.key.deserializer")
    private String keyDescrialzer;


    @Value(value = "zhouxl.kafka.value.deserializer")
    private String valueDescrialzer;


    @Value(value = "zhouxl.kafka.topic")
    private String topic;

    private KafkaProducer<String, String> kafkaProducer;

    public void init() {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);//kafka地址，多个地址用逗号分割
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keyDescrialzer);
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueDescrialzer);
        kafkaProducer = new KafkaProducer<>(p);
    }

    public void sendMsg(int messageNumber) {
        init();
        for (int i = 0; i < messageNumber; i++) {
            String msg = JSONObject.toJSONString(new User(i + 1, "zhouxl" + i, "zhouxl" + i));
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, msg);
            kafkaProducer.send(record);
        }
    }

}
